package com.chenzhiling.study;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import io.delta.flink.sink.DeltaSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.logical.RowType;

/**
 * Author: CHEN ZHI LING
 * Date: 2022/11/17
 * Description:
 */
public class MysqlFlinkDelta {

    public static void main(String[] args) throws Exception {

        String lakePathNoPartition = "lakePath";

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);

        MySqlSource<String> source = MySqlSource
                .<String>builder()
                .hostname("ip")
                .port(3306)
                .databaseList("database")
                .tableList("database.table")
                .username("username")
                .password("password")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();

        RowType rowType = FlinkDeltaUtil.getMysqlRowType();

        DeltaSink<RowData> deltaSink = FlinkDeltaUtil.createDeltaSink(lakePathNoPartition, rowType);

        env.fromSource(source, WatermarkStrategy.noWatermarks(),"demo-mysql-cdc")
                .setParallelism(2)
                //将json数据转为FlinkRowData
                .map(FlinkDeltaUtil::mysqlJsonToRowData)
                .sinkTo(deltaSink)
                .setParallelism(1);

        env.execute("flink-cdc-to-delta");
    }
}
